/**
 * 
 */
package edu.umd.clip.lm.util;

import java.util.concurrent.*;
import java.util.concurrent.locks.*;

/**
 * @author Denis Filimonov <den@cs.umd.edu>
 *
 * Blocks on poll()
 * 
 */
public class ConcurrentQueue<E> extends ConcurrentLinkedQueue<E> {
	/**
	 * 
	 */
	private static final long serialVersionUID = -3630181719779096855L;
	private final Lock lock;
	private final Condition condition;
	private volatile boolean wakeMeUp;
	
	/**
	 * 
	 */
	public ConcurrentQueue() {
		super();
		lock = new ReentrantLock();
		condition = lock.newCondition();
		wakeMeUp = false;
	}

	/* (non-Javadoc)
	 * @see java.util.Queue#add(java.lang.Object)
	 */
	@Override
	public boolean add(E e) {
		boolean result = super.add(e);
		if (wakeMeUp) {
			lock.lock();
			condition.signal();
			lock.unlock();
		}
		return result;
	}

	/* (non-Javadoc)
	 * @see java.util.Queue#offer(java.lang.Object)
	 */
	@Override
	public boolean offer(E e) {
		boolean result = super.offer(e);
		if (wakeMeUp) {
			lock.lock();
			condition.signal();
			lock.unlock();
		}
		return result;
	}

	/* (non-Javadoc)
	 * @see java.util.Queue#poll()
	 */
	@Override
	public E poll() {
		E e = super.poll();
		if (e == null) {
			lock.lock();
			wakeMeUp = true;
			try {
				while(!Thread.currentThread().isInterrupted() && (e = super.poll()) == null) {
					condition.await();
				}
			} catch (InterruptedException e1) {
			}
			wakeMeUp = false;
			lock.unlock();
		}
		return e;
	}
}
